package org.sdnstproject.interlayer.DummyProxyV10;

import com.google.common.primitives.Longs;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;

import org.sdnstproject.interlayer.OF10Protocal.*;
import org.sdnstproject.interlayer.interlayerProxy.*;
import org.sdnstproject.interlayer.utils.Utils;
import org.projectfloodlight.openflow.exceptions.OFParseError;
import org.projectfloodlight.openflow.protocol.*;
import org.projectfloodlight.openflow.protocol.errormsg.OFErrorMsgs;
import org.projectfloodlight.openflow.types.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;

public class DummySwitchV10 extends Thread {
    private static final Logger log = LoggerFactory.getLogger(DummySwitchV10.class);
    public static final int MINIMUM_LENGTH = 8;
    public Socket socket;
    private InputStream in;
    private OutputStream out;
    /* for OF message */
    private OFFactory factory = OFFactories.getFactory(OFVersion.OF_10);;
    private OFMessageReader<OFMessage> reader = factory.getReader();
    private static final DatapathId DEFAULT_DPID = DatapathId.of((long) 1);
    private long requestXid = 0xeeeeeeeel;
    public volatile long responseXid = 0x0L;
    private volatile boolean handshaked = false;
    private DatapathId dpid;
    private Configuration config = Configuration.getInstance();
    private Strategy strategy;
    public volatile boolean running = true;
    public volatile boolean strategySend = false;
    private OFFlowAdd backupFlowAdd;
    private ConcurrentHashMap<OFFlowStatsEntry, OFFlowStatsEntry> flowTable = new ConcurrentHashMap<>();
    private OFState state = new OFState();
    private ResultAnalyzer analyzer = ResultAnalyzer.getInstance();
    private HashSet<OFType> needSynMsg = new HashSet<OFType>();

    public DummySwitchV10() {
        this.dpid = DEFAULT_DPID;
        addSynMsg();
    }
    public DummySwitchV10(Strategy strategy) {
        this.strategy = strategy;
        this.dpid = DEFAULT_DPID;
        addSynMsg();
    }
    public DummySwitchV10(DatapathId dpid) {
        this.dpid = dpid;
        addSynMsg();
    }

    public void notifyOFFuzzer() {
        strategySend = true;
        synchronized (this) {
            notify();
        }
    }

    @Override
    public void run() {
        do {
            try {
                socket = new Socket(config.getControllerIP(), Integer.parseInt(config.getControllerPort()));
                socket.setReuseAddress(true);
            } catch (UnknownHostException e) {
                e.printStackTrace();
                break;
            } catch (SocketException e) {
                log.info(e.toString()+"\n"+Arrays.toString(e.getStackTrace()));
                continue;
            }catch (IOException e) {
                log.info(e.toString()+"\n"+Arrays.toString(e.getStackTrace()));
                continue;
            }
        }while(socket == null);

        try {
            in = socket.getInputStream();
            out = socket.getOutputStream();
        } catch (IOException e2) {
            e2.printStackTrace();
        }

        //发送hello报文
        state.processSendHello();

        byte[] recv;
        int readlen;

        try {
            while (running && !Thread.currentThread().isInterrupted()) {
                recv = new byte[2048];
                if ((readlen = in.read(recv, 0, recv.length)) != -1) {
                    if (readlen >= 8)
                        parseOFMsg(recv, readlen);
                } else
                    break; // end of connection
            }
        } catch (SocketException e) {
            log.info("[DummySwitch] catch SocketException: controller require connection reset");
        }catch (Exception e) {
            // if any error occurs
            Thread.currentThread().interrupt();
            e.printStackTrace();
        } finally {
            try {
                if (in != null)
                    in.close();
                if (socket != null)
                    socket.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }

    // for OpenFlow Message
    private boolean parseOFMsg(byte[] recv, int len) {
        byte[] rawMsg = new byte[len];
        System.arraycopy(recv, 0, rawMsg, 0, len);
        ByteBuf bb = Unpooled.copiedBuffer(rawMsg);

        int totalLen = bb.readableBytes();
        int offset = bb.readerIndex();

        while (offset < totalLen) {
            bb.readerIndex(offset);

            byte version = bb.readByte();

            if (version != 1) {
                log.info("[Channel Agent] OF Version >= 1.4 not supported");
                return false;
            }

            // version check
            if (version != factory.getVersion().getWireVersion()) {
                log.info("[Channel Agent] Received OF Version {} not matched with the dummy switch", version);
                return false;
            }

            bb.readByte();
            int length = U16.f(bb.readShort());
            bb.readerIndex(offset);

            if (length < MINIMUM_LENGTH) {
                log.info("Wrong length: Expected to be >= " + MINIMUM_LENGTH + ", was: " + length);
                return false;
            }

            OFMessage message = null;
            try {
                message = reader.readFrom(bb);
            } catch (OFParseError e) {
                analyzer.isReceiveErrorMsg = true;
                log.info(e.toString());
                notifyOFFuzzer();
                return true;
            }
            responseXid = message.getXid();
            //if get correct response xid, then notify the state.
            if(responseXid == requestXid) {
                synchronized (state) {
                    state.notify();
                }
            }else if(strategy.field.equals("xid")) {
                if(Long.parseLong(strategy.actionValue) == responseXid) {
                    synchronized (state) {
                        state.notify();
                    }
                }
            }
            log.info(message.toString());
            state.processOFMsg(message);
            offset += length;
        }

        bb.clear();
        return true;
    }

    public class OFState extends Thread{
        void processOFMsg(OFMessage m) {
            switch(m.getType()) {
                case HELLO:
                    processOFHello((OFHello)m);
                    break;
                case FEATURES_REQUEST:
                    processOFFeaturesReq((OFFeaturesRequest)m);
                    break;
                case GET_CONFIG_REQUEST:
                    processOFGetConfigRequest((OFGetConfigRequest)m);
                    break;
                case QUEUE_GET_CONFIG_REQUEST:
                    processOFQueueGetConfigRequest((OFQueueGetConfigRequest)m);
                    break;
                case SET_CONFIG:
                    processOFSetConfig((OFSetConfig)m);
                    break;
                case BARRIER_REQUEST:
                    processOFBarrierRequest((OFBarrierRequest)m);
                    break;
                case STATS_REQUEST:
                    processOFStatsRequest((OFStatsRequest)m);
                    break;
                case EXPERIMENTER:
                    processOFExperimenter((OFExperimenter)m);
                    break;
                case ECHO_REQUEST:
                    processOFEchoRequest((OFEchoRequest)m);
                    break;
                case ECHO_REPLY:
                    processOFEchoReply((OFEchoReply)m);
                    break;
                case ERROR:
                    processOFError((OFErrorMsg)m);
                    break;
                case FLOW_MOD:
                    processOFFlowMod((OFFlowMod)m);
                    break;
                case PORT_MOD:
                    processOFPortMod((OFPortMod)m);
                    break;
                case PACKET_OUT:
                    processOFPacketOut((OFPacketOut)m);
                    break;
                default:
                    log.error("unkown message: " + m.getType());
                    break;
            }
        }

        private void processOFPortMod(OFPortMod m) {
            log.info("Reveived a OFPortMod.");
        }

        private void processOFSetConfig(OFSetConfig m) {
            log.info("Reveived a OFSetConfig.");
        }

        private void processOFQueueGetConfigRequest(OFQueueGetConfigRequest m) {
            log.info("Reveived a OFQueueGetConfigRequest.");
        }

        void processOFPacketOut(OFPacketOut m) {
            log.info("Reveived a OFPacketOut.");
        }

        void processOFHello(OFHello m) {
            log.info("Reveived a OFHello.");
        }

        public void processSendHello() {
            if(strategy != null && strategy.ofType == OFType.HELLO) {
                switch (strategy.action) {
                    case MODIFY:
                        FuHelloV10 hello = new FuHelloV10();
                        hello.setField(strategy.field, strategy.actionValue);
                        sendByteBufMsg(hello.writeTo());
                        notifyOFFuzzer();
                        log.info("send a HELLO msg to controller.");
                        break;
                    case DELAY:
                        try {
                            Thread.sleep(Long.parseLong(strategy.actionValue));
                        } catch (Exception e) {
                            log.info(e.toString());
                        }
                        sendHello(0);
                        notifyOFFuzzer();
                        log.info("send a HELLO msg to controller.");
                        break;
                    case DUPLICATE:
                        for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                            sendHello(0);
                        }
                        notifyOFFuzzer();
                        log.info("send "+strategy.actionValue+" HELLO msg to controller.");
                        break;
                    case DROP:
                        notifyOFFuzzer();
                        return;
                }
            }else {
                sendHello(0);
                log.info("send a HELLO msg to controller.");
            }
        }

        void processOFFlowMod(OFFlowMod m) {
            log.info("Reveived a OFFlowMod.");
            OFFlowMod fm = (OFFlowMod) m;

            if (fm.getCommand() == OFFlowModCommand.ADD) {
                OFFlowAdd fa = (OFFlowAdd) fm;
                if (fa.getPriority() == 555) {
                    backupFlowAdd = fa;
                    log.info("[Channel Agent] catch un-flagged msg {}", fa.toString());
                }
            }
        }

        void processOFError(OFErrorMsg m) {
            log.info("Reveived a OFErrorMsg.");
        }

        void processOFEchoReply(OFEchoReply m) {
            log.info("Reveived a OFEchoReply.");
        }

        void processOFEchoRequest(OFEchoRequest m) {
            log.info("Reveived a OFEchoRequest.");
            if(strategy != null && strategy.ofType == OFType.ECHO_REPLY) {
                switch (strategy.action) {
                    case MODIFY:
                        FuEchoReplyV10 echoReply = new FuEchoReplyV10();
                        echoReply.xid = m.getXid();
                        echoReply.setField(strategy.field, strategy.actionValue);
                        sendByteBufMsg(echoReply.writeTo());
                        notifyOFFuzzer();
                        log.info("send a echoReply msg to controller.");
                        break;
                    case DELAY:
                        try {
                            Thread.sleep(Long.parseLong(strategy.actionValue));
                        } catch (Exception e) {
                            log.info(e.toString());
                        }
                        sendEchoReply(m.getXid());
                        notifyOFFuzzer();
                        log.info("send a echoReply msg to controller.");
                        break;
                    case DUPLICATE:
                        for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                            sendEchoReply(m.getXid());
                        }
                        notifyOFFuzzer();
                        log.info("send "+strategy.actionValue+" echoReply msg to controller.");
                        break;
                    case DROP:
                        notifyOFFuzzer();
                        return;
                }
            }else {
                sendEchoReply(m.getXid());
                log.info("send a echoReply msg to controller.");
//				byte[] msg = Utils.hexStringToByteArray(DummyOF13.ECHO_REPLY);
//        		sendRawMsg(msg);
            }
        }

        void processOFStatsRequest(OFStatsRequest m) {
            log.info("Reveived a OFStatsRequest.");
            sendStatReply(m, m.getXid());
//            handshaked = true;
            log.info("send a StatsReply msg to controller.");
        }

        void processOFBarrierRequest(OFBarrierRequest m) {
            log.info("Reveived a OFBarrierRequest.");
            if(strategy != null && strategy.ofType == OFType.BARRIER_REPLY) {
                switch (strategy.action) {
                    case MODIFY:
                        FuBarrierReplyV10 barrierReply = new FuBarrierReplyV10();
                        barrierReply.xid = m.getXid();
                        barrierReply.setField(strategy.field, strategy.actionValue);
                        sendByteBufMsg(barrierReply.writeTo());
                        notifyOFFuzzer();
                        log.info("send a BarrierReply msg to controller.");
                        break;
                    case DELAY:
                        try {
                            Thread.sleep(Long.parseLong(strategy.actionValue));
                        } catch (Exception e) {
                            log.info(e.toString());
                        }
                        sendBarrierRes(m.getXid());
                        notifyOFFuzzer();
                        log.info("send a BarrierReply msg to controller.");
                        break;
                    case DUPLICATE:
                        for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                            sendBarrierRes(m.getXid());
                        }
                        notifyOFFuzzer();
                        log.info("send "+strategy.actionValue+" BarrierReply msg to controller.");
                        break;
                    case DROP:
                        notifyOFFuzzer();
                        return;
                }
            }else {
                sendBarrierRes(m.getXid());
                log.info("send a BarrierReply msg to controller.");
            }
        }

        void processOFExperimenter(OFExperimenter m) {
            log.info("Reveived a OFExperimenter.");
            sendExperimenter(m.getXid());
            log.info("send a ExperimenterReply msg to controller.");
        }

        void processOFGetConfigRequest(OFGetConfigRequest m) {
            log.info("Reveived a OFGetConfigRequest.");
            if(strategy != null && strategy.ofType == OFType.GET_CONFIG_REPLY) {
                switch (strategy.action) {
                    case MODIFY:
                        FuGetConfigReplyV10 getConfigReply = new FuGetConfigReplyV10();
                        getConfigReply.xid = m.getXid();
                        getConfigReply.setField(strategy.field, strategy.actionValue);
                        sendByteBufMsg(getConfigReply.writeTo());
                        notifyOFFuzzer();
                        log.info("send a ConfigReply msg to controller.");
                        break;
                    case DELAY:
                        try {
                            Thread.sleep(Long.parseLong(strategy.actionValue));
                        } catch (Exception e) {
                            log.info(e.toString());
                        }
                        sendGetConfigReply(m.getXid());
                        notifyOFFuzzer();
                        log.info("send a ConfigReply msg to controller.");
                        break;
                    case DUPLICATE:
                        for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                            sendGetConfigReply(m.getXid());
                        }
                        notifyOFFuzzer();
                        log.info("send "+strategy.actionValue+" ConfigReply msg to controller.");
                        break;
                    case DROP:
                        notifyOFFuzzer();
                        return;
                }
            }else {
                sendGetConfigReply(m.getXid());
                log.info("send a ConfigReply msg to controller.");
            }
        }

        void processOFFeaturesReq(OFFeaturesRequest m) {
            log.info("receive OFFeaturesRequest msg"+m.toString());
            //send FeatureReply
            if(strategy != null && strategy.ofType == OFType.FEATURES_REPLY) {
                switch (strategy.action) {
                    case MODIFY:
                        FuFeaturesReplyV10 featuresReply = new FuFeaturesReplyV10();
                        featuresReply.xid = m.getXid();
                        featuresReply.datapathId = dpid.getLong();
                        featuresReply.setField(strategy.field, strategy.actionValue);
                        sendByteBufMsg(featuresReply.writeTo());
                        notifyOFFuzzer();
                        log.info("send a featuresReply msg to controller.");
                        break;
                    case DELAY:
                        try {
                            Thread.sleep(Long.parseLong(strategy.actionValue));
                        } catch (Exception e) {
                            log.info(e.toString());
                        }
                        try {
                            sendFeatureReply(m.getXid());
                        } catch (OFParseError e) {
                            e.printStackTrace();
                        }
                        notifyOFFuzzer();
                        log.info("send a featuresReply msg to controller.");
                        break;
                    case DUPLICATE:
                        for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                            try {
                                sendFeatureReply(m.getXid());
                            } catch (OFParseError e) {
                                e.printStackTrace();
                            }
                        }
                        notifyOFFuzzer();
                        log.info("send "+strategy.actionValue+" featuresReply msg to controller.");
                        break;
                    case DROP:
                        notifyOFFuzzer();
                        return;
                }
            }else {
//        		byte[] msg = Utils.hexStringToByteArray(DummyOF13.FEATURE_REPLY);
//        		sendRawMsg(msg);
//        		ByteBuf bb =  Unpooled.copiedBuffer(msg);
//        		OFMessage msgMessage = null;
//        		try {
//        			msgMessage = OFFactories.getFactory(OFVersion.OF_13).getReader().readFrom(bb);
//        		} catch (OFParseError e1) {
//        			e1.printStackTrace();
//        		}
//        		OFFeaturesReply reply = (OFFeaturesReply)msgMessage;

                try {
                    sendFeatureReply(m.getXid());
                    log.info("send a featuresReply msg to controller.");
                } catch (OFParseError e) {
                    e.printStackTrace();
                }
            }
            handshaked = true;
            if(strategy != null && needSynMsg.contains(strategy.ofType)) {
                state.start();
                log.info("asynchronous state machine has start.");
            }

        }

        @Override
        public void run() {
            switch (strategy.ofType) {
                case PACKET_IN:
                    sendPACKET_INtoController();
                    break;
                case ECHO_REQUEST:
                    sendECHO_REQUESTtoController();
                    break;
                case FLOW_REMOVED:
                    sendFLOW_REMOVEDtoController();
                    break;
                case ERROR:
                    sendERRORtoController();
                    break;
                case PORT_STATUS:
                    sendPORT_STATUStoController();
                    break;
            }
            strategySend = true;
            //after send syn stratege msg. waitting 3sec to get response msg or waitting to been notified.
            synchronized (this) {
                if(responseXid != requestXid) {
                    try {
                        wait(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            if(responseXid == requestXid) {
                log.info("responseXid == requestXid");
                analyzer.notReceiveResponseMsg = false;
            }else {
                log.info("responseXid != requestXid");
                analyzer.notReceiveResponseMsg = true;
            }
            //notify ResultAnalyzer to get result.
            notifyOFFuzzer();
        }
    }

    public void sendPACKET_INtoController() {
        switch (strategy.action) {
            case MODIFY:
                FuPacketInV10 packetIn = new FuPacketInV10();
                packetIn.xid = requestXid;
                packetIn.setField(strategy.field, strategy.actionValue);
                sendByteBufMsg(packetIn.writeTo());
                log.info("state send a PACKET_IN packet to Controller.");
                break;
            case DELAY:
                try {
                    Thread.sleep(Long.parseLong(strategy.actionValue));
                } catch (Exception e) {
                    log.info(e.toString());
                }
                sendByteBufMsg(new FuPacketInV10().setXid(requestXid).writeTo());
                log.info("state send a PACKET_IN packet to Controller.");
                break;
            case DUPLICATE:
                for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                    sendByteBufMsg(new FuPacketInV10().setXid(requestXid).writeTo());
                }
                log.info("state send "+strategy.actionValue+" PACKET_IN packet to Controller.");
                break;
            case DROP:
                return;
        }
    }

    public void sendPORT_STATUStoController() {
        switch (strategy.action) {
            case MODIFY:
                FuPortStatusV10 portStatus = new FuPortStatusV10();
                portStatus.xid = requestXid;
                portStatus.setField(strategy.field, strategy.actionValue);
                sendByteBufMsg(portStatus.writeTo());
                log.info("state send a PORT_STATUS packet to Controller.");
                break;
            case DELAY:
                try {
                    Thread.sleep(Long.parseLong(strategy.actionValue));
                } catch (Exception e) {
                    log.info(e.toString());
                }
                sendByteBufMsg(new FuPortStatusV10().setXid(requestXid).writeTo());
                log.info("state send a PORT_STATUS packet to Controller.");
                break;
            case DUPLICATE:
                for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                    sendByteBufMsg(new FuPortStatusV10().setXid(requestXid).writeTo());
                }
                log.info("state send "+strategy.actionValue+" PORT_STATUS packet to Controller.");
                break;
            case DROP:
                return;
        }
    }

    public void sendERRORtoController() {
        switch (strategy.action) {
            case MODIFY:
                FuErrorV10 error = new FuErrorV10();
                error.xid = requestXid;
                error.setField(strategy.field, strategy.actionValue);
                sendByteBufMsg(error.writeTo());
                log.info("state send a ERROR packet to Controller.");
                break;
            case DELAY:
                try {
                    Thread.sleep(Long.parseLong(strategy.actionValue));
                } catch (Exception e) {
                    log.info(e.toString());
                }
                sendByteBufMsg(new FuErrorV10().setXid(requestXid).writeTo());
                log.info("state send a ERROR packet to Controller.");
                break;
            case DUPLICATE:
                for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                    sendByteBufMsg(new FuErrorV10().setXid(requestXid).writeTo());
                }
                log.info("state send "+strategy.actionValue+" ERROR packet to Controller.");
                break;
            case DROP:
                return;
        }
    }

    public void sendFLOW_REMOVEDtoController() {
        switch (strategy.action) {
            case MODIFY:
                FuFlowRemovedV10 flowRemoved = new FuFlowRemovedV10();
                flowRemoved.xid = requestXid;
                flowRemoved.setField(strategy.field, strategy.actionValue);
                sendByteBufMsg(flowRemoved.writeTo());
                log.info("state send a FLOW_REMOVED packet to Controller.");
                break;
            case DELAY:
                try {
                    Thread.sleep(Long.parseLong(strategy.actionValue));
                } catch (Exception e) {
                    log.info(e.toString());
                }
                sendByteBufMsg(new FuFlowRemovedV10().setXid(requestXid).writeTo());
                log.info("state send a FLOW_REMOVED packet to Controller.");
                break;
            case DUPLICATE:
                for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                    sendByteBufMsg(new FuFlowRemovedV10().setXid(requestXid).writeTo());
                }
                log.info("state send "+strategy.actionValue+" FLOW_REMOVED packet to Controller.");
                break;
            case DROP:
                return;
        }
    }

    public void sendECHO_REQUESTtoController() {
        switch (strategy.action) {
            case MODIFY:
                FuEchoRequestV10 echoRequest = new FuEchoRequestV10();
                echoRequest.xid = requestXid;
                echoRequest.setField(strategy.field, strategy.actionValue);
                sendByteBufMsg(echoRequest.writeTo());
                log.info("state send a ECHO_REQUEST packet to Controller.");
                notifyOFFuzzer();
                break;
            case DELAY:
                try {
                    Thread.sleep(Long.parseLong(strategy.actionValue));
                } catch (Exception e) {
                    log.info(e.toString());
                }
                sendByteBufMsg(new FuEchoRequestV10().setXid(requestXid).writeTo());
                log.info("state send a ECHO_REQUEST packet to Controller.");
                break;
            case DUPLICATE:
                for(int i = 0; i<Integer.parseInt(strategy.actionValue); i++) {
                    sendByteBufMsg(new FuEchoRequestV10().setXid(requestXid).writeTo());
                }
                log.info("state send "+strategy.actionValue+" ECHO_REQUEST packet to Controller.");
                break;
            case DROP:
                return;
        }
    }

    public void sendMsg(OFMessage msg, int len) {
        ByteBuf buf;

        if (len == -1) {
            buf = PooledByteBufAllocator.DEFAULT.directBuffer(1024);
        } else {
            buf = PooledByteBufAllocator.DEFAULT.directBuffer(len);
        }

        msg.writeTo(buf);

        int length = buf.readableBytes();
        byte[] bytes = new byte[length];
        buf.getBytes(buf.readerIndex(), bytes);

        if(socket.isClosed()) {
            log.info("Sock is closed, could not sendMsg.");
            return;
        }
        try {
            this.out.write(bytes, 0, length);
        } catch (IOException e) {
            e.printStackTrace();
        }

        buf.clear();
        buf.release();
    }

    public void sendRawMsg(byte[] msg) {
        if(socket.isClosed()) {
            log.info("Sock is closed, could not sendMsg.");
            return;
        }
        try {
            this.out.write(msg, 0, msg.length);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void sendByteBufMsg(ByteBuf buf) {
        int length = buf.readableBytes();
        byte[] bytes = new byte[length];
        buf.getBytes(buf.readerIndex(), bytes);

        if(socket.isClosed()) {
            log.info("Sock is closed, could not sendMsg.");
            return;
        }
        try {
            this.out.write(bytes, 0, length);
        } catch (IOException e) {
            e.printStackTrace();
        }

        buf.clear();
        buf.release();
    }

    public OFFlowAdd getBackupFlowAdd() {
        return backupFlowAdd;
    }


    public void sendError() throws OFParseError {
        long r_xid = 0xeeeeeeeel;

        OFErrorMsgs msg = factory.errorMsgs();
        OFFeaturesReply.Builder frb = factory.buildFeaturesReply();
        OFFeaturesReply fr = frb.build();

        return;
    }

    public ByteBuf sendFlowRemoved() throws OFParseError {
        OFFactory factory = OFFactories.getFactory(OFVersion.OF_10);
        long r_xid = 0xeeeeeeeel;

        OFFlowRemoved.Builder fab = factory.buildFlowRemoved();
        fab.setXid(r_xid);
        OFFlowRemoved hello = fab.build();

        ByteBuf buf = null;
        buf = PooledByteBufAllocator.DEFAULT.directBuffer(88);
        hello.writeTo(buf);

        byte[] bytes;
        int length = buf.readableBytes();
        bytes = new byte[length];
        buf.getBytes(buf.readerIndex(), bytes);

        try {
            this.out.write(bytes, 0, length);
        } catch (IOException e) {
            e.printStackTrace();
        }

        return buf;
    }

    public ByteBuf sendPortStatus() throws OFParseError {
        OFFactory factory = OFFactories.getFactory(OFVersion.OF_10);
        long r_xid = 0xeeeeeeeel;

        OFPortStatus.Builder fab = factory.buildPortStatus();
        fab.setXid(r_xid);
        OFPortStatus hello = fab.build();

        ByteBuf buf = null;
        buf = PooledByteBufAllocator.DEFAULT.directBuffer(64);
        hello.writeTo(buf);

        byte[] bytes;
        int length = buf.readableBytes();
        bytes = new byte[length];
        buf.getBytes(buf.readerIndex(), bytes);

        try {
            this.out.write(bytes, 0, length);
        } catch (IOException e) {
            e.printStackTrace();
        }

        return buf;
    }

    /* OF HandShake */
    public void sendHello(long xid){
        OFHello.Builder builder = factory.buildHello();
        builder.setXid(0x1);

        sendMsg(builder.build(), -1);
        log.info("send a of10 hello packet to controller.");
    }

    public void sendFeatureReply(long xid) throws OFParseError {
        OFFeaturesReply.Builder frb = factory.buildFeaturesReply();
        frb.setDatapathId(this.dpid);
        frb.setNBuffers((long) 0);//256
        frb.setNTables((short) 254);
        frb.setXid(xid);
        HashSet<OFCapabilities> ofCapabilities = new HashSet<OFCapabilities>();
        ofCapabilities.add(OFCapabilities.FLOW_STATS);
        ofCapabilities.add(OFCapabilities.TABLE_STATS);
        ofCapabilities.add(OFCapabilities.PORT_STATS);
        ofCapabilities.add(OFCapabilities.QUEUE_STATS);
        frb.setCapabilities(ofCapabilities);
        OFFeaturesReply msg = frb.build();
        log.info("send a FeatureReply: "+msg.toString());
        sendMsg(msg, -1);
    }

    public void sendGetConfigReply(long xid) {
        byte[] msg;
        msg = Utils.hexStringToByteArray(DummyOF10.GET_CONFIG_REPLY);
        byte[] xidbytes = Longs.toByteArray(xid);
        System.arraycopy(xidbytes, 4, msg, 4, 4);
        sendRawMsg(msg);
    }

    public void sendStatReply(OFMessage input, long xid) {
        byte[] msg;
        msg = Utils.hexStringToByteArray(DummyOF10.STATS_REPLY);
        byte[] xidbytes = Longs.toByteArray(xid);
        System.arraycopy(xidbytes, 4, msg, 4, 4);
        sendRawMsg(msg);
    }

    private void sendRoleRes(long xid) {
        OFRoleReply.Builder builder = factory.buildRoleReply();
        builder.setXid(xid);
        builder.setGenerationId(U64.of(0));
        builder.setRole(OFControllerRole.ROLE_MASTER);

        OFRoleReply msg = builder.build();
        sendMsg(msg, -1);
    }

    private void sendExperimenter(long xid) {
        byte[] msg = DummyOF10.hexStringToByteArray(DummyOF10.experimenter);
        byte[] xidbytes = Longs.toByteArray(xid);
        System.arraycopy(xidbytes, 4, msg, 4, 4);
        sendRawMsg(msg);
    }

    private void sendBarrierRes(long xid) {
        OFBarrierReply.Builder builder = factory.buildBarrierReply();
        builder.setXid(xid);
        OFBarrierReply msg = builder.build();
        sendMsg(msg, 8);
    }

    private void sendEchoReply(long xid) {
        OFEchoReply.Builder builder = factory.buildEchoReply();
        builder.setXid(xid);
        OFEchoReply msg = builder.build();
        sendMsg(msg, -1);
    }

    private void addSynMsg() {
        needSynMsg.add(OFType.ECHO_REQUEST);
        needSynMsg.add(OFType.PACKET_IN);
        needSynMsg.add(OFType.PORT_STATUS);
        needSynMsg.add(OFType.ERROR);
        needSynMsg.add(OFType.FLOW_REMOVED);
    }

}
